package atguigu.com.edu.app.dws;

import atguigu.com.edu.bean.TrafficPageViewBean;
import atguigu.com.edu.util.DateFormatUtil;
import atguigu.com.edu.util.MyClickhouseUtil;
import atguigu.com.edu.util.MyKafkaUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;


import java.time.Duration;


public class DwsTrafficVcChArIsNewPageViewWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(4);
        //设置检查点
      /*  env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        //设置检查点超时时间
        env.getCheckpointConfig().setCheckpointTimeout(6000L);
        //设置job取消，检查点是否保留
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //设置两个检查点最小时间间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        //设置重启
       env.setRestartStrategy(RestartStrategies.failureRateRestart(3,Time.days(30), Time.seconds(3)));

      env.setStateBackend(new HashMapStateBackend());

      env.getCheckpointConfig().setCheckpointStorage(" ");

      System.setProperty(""," ");
      */
        String pageLogTopic = "dwd_traffic_page_log";
        String uvTopic = "dwd_traffic_unique_visitor_detail";
        String ujdTopic = "dwd_traffic_user_jump_detail";
        String groupId = "dws_traffic_vc_ch_ar_isnew_group";

        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(pageLogTopic, groupId);

        FlinkKafkaConsumer<String> kafkaConsumeruv = MyKafkaUtil.getKafkaConsumer(uvTopic, groupId);
        FlinkKafkaConsumer<String> kafkaConsumeruj = MyKafkaUtil.getKafkaConsumer(ujdTopic, groupId);

        DataStreamSource<String> pagelog = env.addSource(kafkaConsumer);
        DataStreamSource<String> uv = env.addSource(kafkaConsumeruv);
        DataStreamSource<String> ujsource = env.addSource(kafkaConsumeruj);

      /*  pagelog.print("$$$$$$");
        uv.print(">>>>>");
        ujsource.print("???????");
        */
        //对读取数据进行类型转换
        SingleOutputStreamOperator<TrafficPageViewBean> papelogStatsDs = pagelog.map(
                new MapFunction<String, TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean map(String jsonStr) throws Exception {
                        //为了方便进行处理，将json字符串转换为json对象
                        JSONObject jsonObj = JSON.parseObject(jsonStr);
                        JSONObject commonJsonObj = jsonObj.getJSONObject("common");
                        JSONObject pageJsonObj = jsonObj.getJSONObject("page");
                        Long ts = jsonObj.getLong("ts");
                        TrafficPageViewBean pageViewBean = null;

                        pageViewBean = new TrafficPageViewBean(
                                "",
                                "",
                                commonJsonObj.getString("vc"),
                                commonJsonObj.getString("ch"),
                                commonJsonObj.getString("ar"),
                                commonJsonObj.getString("is_new"),
                                0L,
                                1L,
                                1L,
                                1L,
                                0L,
                                ts
                        );

                        return pageViewBean;
                    }
                }
        );
        SingleOutputStreamOperator<TrafficPageViewBean> uvStatsDs = uv.map(new MapFunction<String, TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean map(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                JSONObject common = jsonObject.getJSONObject("common");
                Long ts = jsonObject.getLong("ts");
                TrafficPageViewBean PageViewBean = new TrafficPageViewBean(
                        "",
                        "",
                        common.getString("vc"),
                        common.getString("ch"),
                        common.getString("ar"),
                        common.getString("is_new"),
                        1L,
                        0L,
                        0L,
                        0L,
                        0L,
                        ts

                );
                return PageViewBean;
            }
        });
        SingleOutputStreamOperator<TrafficPageViewBean> ujStatsDs = ujsource.map(new MapFunction<String, TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean map(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);

                JSONObject common = jsonObject.getJSONObject("common");

                Long ts = jsonObject.getLong("ts");
                TrafficPageViewBean trafficPageViewBean = new TrafficPageViewBean(
                        "",
                        "",
                        common.getString("vc"),
                        common.getString("ch"),
                        common.getString("ar"),
                        common.getString("is_new"),
                        0L,
                        0L,
                        0L,
                        0L,
                        1L,
                        ts
                );
                return trafficPageViewBean;

            }
        });

        DataStream<TrafficPageViewBean> unionsDs = papelogStatsDs.union(
                ujStatsDs, uvStatsDs
        );

        //unionsDs.print(">>>>>");

        SingleOutputStreamOperator<TrafficPageViewBean> withWatermarkDs = unionsDs.assignTimestampsAndWatermarks(WatermarkStrategy.<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner(
                new SerializableTimestampAssigner<TrafficPageViewBean>() {
                    @Override
                    public long extractTimestamp(TrafficPageViewBean trafficPageViewBean, long l) {
                        return trafficPageViewBean.getTs();
                    }
                }
                )
        );

        KeyedStream<TrafficPageViewBean, Tuple4<String, String, String, String>> keydDs = withWatermarkDs.keyBy(new KeySelector<TrafficPageViewBean, Tuple4<String, String, String, String>>() {
            @Override
            public Tuple4<String, String, String, String> getKey(TrafficPageViewBean trafficPageViewBean) throws Exception {
                return Tuple4.of(trafficPageViewBean.getVc(),
                        trafficPageViewBean.getCh(),
                        trafficPageViewBean.getAr(),
                        trafficPageViewBean.getIsNew()
                );
            }
        });

        WindowedStream<TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow> window = keydDs.window(TumblingEventTimeWindows.of(Time.seconds(10)));

        SingleOutputStreamOperator<TrafficPageViewBean> reduce = window
                .reduce(new ReduceFunction<TrafficPageViewBean>() {
                            @Override
                            public TrafficPageViewBean reduce(TrafficPageViewBean t, TrafficPageViewBean t1) throws Exception {
                                t.setPvCt(t.getPvCt() + t1.getPvCt());
                                t.setSvCt(t.getSvCt() + t1.getSvCt());
                                t.setDurSum(t.getDurSum() + t1.getDurSum());
                                t.setUvCt(t.getUvCt() + t1.getUvCt());
                                t.setUjCt(t.getUjCt() + t1.getUjCt());
                                return t;

                            }
                        }, new WindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow>() {
                            @Override
                            public void apply(Tuple4<String, String, String, String> stringStringStringStringTuple4, TimeWindow timeWindow,
                                              Iterable<TrafficPageViewBean> iterable, Collector<TrafficPageViewBean> collector) throws Exception {
                                String stt = DateFormatUtil.toYmdHms(timeWindow.getStart());
                                String edt = DateFormatUtil.toYmdHms(timeWindow.getEnd());
                                for (
                                        TrafficPageViewBean viewBean : iterable) {
                                    viewBean.setStt(stt);
                                    viewBean.setEdt(edt);
                                    viewBean.setTs(System.currentTimeMillis());
                                    collector.collect(viewBean);
                                }
                            }
                        }
                );

        reduce.print(">>>>>");


      reduce.addSink(MyClickhouseUtil.getSinkFunction("insert into dws_traffic_vc_ch_ar_is_new_page_view_window values(?,?,?,?,?,?,?,?,?,?,?,?)"));

        env.execute();


    }
}
